package com.tencent.sr.iris.activity.service.kafka.listener;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.TypeReference;
import com.google.common.util.concurrent.RateLimiter;
import com.tencent.sr.iris.activity.common.config.DependencyConfig;
import com.tencent.sr.iris.activity.repository.entity.TIrisUserTaskRecordDO;
import com.tencent.sr.iris.activity.service.activity.impl.IrisBatchDataServiceImpl;
import com.tencent.sr.iris.activity.service.config.ActivityConfig;
import com.tencent.sr.iris.activity.service.event.dto.SyncEventDTO;
import com.tencent.sr.rmall.springbootstarter.kafka.KafkaMetaMessage;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.DependsOn;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * @author nixiaolin
 * @date 2023/9/5 9:54
 */
@Slf4j
@Component
@DependsOn("kafkaParamConfiguration")
public class BatchDataSyncListener {

    @Resource
    private IrisBatchDataServiceImpl irisBatchDataService;

    @Resource
    private DependencyConfig dependencyConfig;

    @Resource
    private ActivityConfig activityConfig;

    private static final RateLimiter RATE_LIMITER = RateLimiter.create(10);

    @KafkaListener(topics = "${kafka.topics.activityTaskRecordToBatchDataTopic}",
            groupId = "${retail.kafka.consumer.groupId.irisActivityService}",
            containerFactory = "kafkaListenerContainerFactory")
    public void activityTaskRecordToBatchData(KafkaMetaMessage kafkaMetaMessage, Acknowledgment ack) {
        if (activityConfig.isSyncBatchDataStop()) {
            ack.acknowledge();
            log.info("activityTaskRecordToBatchData.stop msg: {}", JSON.toJSONString(kafkaMetaMessage));
            return;
        }

        log.info("activityTaskRecordToBatchData.handle msg: {}", JSON.toJSONString(kafkaMetaMessage));
        try {
            DependencyConfig.Config config = dependencyConfig.getConfigs().get("syncBatchData");
            if (config != null && config.getPermitsPerSecond() != null) {
                RATE_LIMITER.setRate(config.getPermitsPerSecond());
                RATE_LIMITER.acquire();
                log.info("跑批数据同步，限流{}qps");
            }
            String jsonString = String.valueOf(kafkaMetaMessage.getData());
            SyncEventDTO<TIrisUserTaskRecordDO> request = JSONObject.parseObject(jsonString,
                    new TypeReference<SyncEventDTO<TIrisUserTaskRecordDO>>() {
                    });

            irisBatchDataService.handleBatchDataSync(request);
        } catch (Exception e) {
            log.error("activityTaskRecordToBatchData消息处理异常，kafkaMetaMessage:{}", JSON.toJSONString(kafkaMetaMessage), e);
        }
        ack.acknowledge();
    }
}
